package com.voice.core.channel.tcp;

import javax.xml.ws.WebServiceException;
import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.*;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

public class IOProcessThread extends Thread implements Closeable {

    private AtomicBoolean isClosed=new AtomicBoolean(false);
    private AtomicBoolean inRegister=new AtomicBoolean(false);
    private Selector selector;
    private int ops;
    private IOProcessEvent ioProcessEvent;

    public IOProcessThread(String name,int ops,IOProcessEvent ioProcessEvent) throws IOException {
        super(name);
        this.selector=Selector.open();
        this.ops=ops;
        this.ioProcessEvent=ioProcessEvent;
    }

    @Override
    public void run() {
        while (!isClosed.get()) {
            try {
                if (selector.select() == 0){
                    synchronized (inRegister){
                        if(inRegister.get()) inRegister.wait();
                    }
                    continue;
                }
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                selectionKeys.stream().forEach(selectionKey -> {
                    if(!selectionKey.isValid()) return;
                    selectionKey.interestOps(selectionKey.readyOps() & ~ops);
                    if(ioProcessEvent!=null)ioProcessEvent.onEventProcess((SocketChannel) selectionKey.channel(),(Runnable) selectionKey.attachment());
                });
                selectionKeys.clear();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public SelectionKey register(SocketChannel socketChannel,Runnable runnable){
        SelectionKey selectionKey = null;
        synchronized (inRegister){
            inRegister.set(true);
            selector.wakeup();
            try {
                if (socketChannel.isRegistered()) {
                    selectionKey = socketChannel.keyFor(selector);
                    if(selectionKey!=null&&selectionKey.isValid()){
                        selectionKey.interestOps(selectionKey.readyOps() | ops);
                    }
                }
                if (selectionKey == null) selectionKey = socketChannel.register(selector, ops,runnable);
            } catch (ClosedChannelException e) {}
            inRegister.set(false);
        }
        return selectionKey;
    }

    public void unRegister(SocketChannel socketChannel) {
        if (socketChannel.isRegistered()) {
            SelectionKey selectionKey = socketChannel.keyFor(selector);
            if (selectionKey != null) {
                selectionKey.cancel();
                selector.wakeup();
            }
        }
    }

    @Override
    public void close() throws WebServiceException {
        isClosed.set(true);
        selector.wakeup();
    }

    public interface IOProcessEvent{
        void onEventProcess(SocketChannel channel, Runnable runnable);
    }
}
